package org.apache.ocean.solr.replication;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.sourceforge.jsorter.ReflectColumns;
import net.sourceforge.jsorter.SortReflect;
import net.sourceforge.jsorter.Sorter;
import net.sourceforge.jsorter.SorterConstants;
import net.sourceforge.jsorter.SortReflect.MethodException;

import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.RequestEntity;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.lucene.store.RAMDirectory;
import org.jdom.Element;

// TODO: reset sequences on master change
public class Replication {
	public static Logger log = Logger.getLogger(Replication.class.getName());
	public static long LEASE_DURATION = 1000 * 5;
	private String id;
	private URL url;
	private LogFileManager logFileManager;
	private File replicationDirectoryFile;
	private LongSequence sequence;
	private Loader loader;
	private ReentrantLock writeLock = new ReentrantLock();
	private ByteBufferPool documentsBytePool = new ByteBufferPool(50 * 1024, 5, 5);
	private ByteBufferPool deletesBytePool = new ByteBufferPool(5 * 1024, 5, 10);
	private Set<URL> nodeUrls = new HashSet<URL>();
	private String operatingRole;
	private Node chosenMaster;
	private ExecutorService threadPool;
	private Leases leases;
	private Client client;
	private NameInterface nameInterface;
	private int serverNumber;

	public Replication(String id, int serverNumber, URL url, File replicationDirectoryFile, Loader loader) throws Exception, IOException {
		this.id = id;
		this.serverNumber = serverNumber;
		this.url = url;
		this.replicationDirectoryFile = replicationDirectoryFile;
		this.loader = loader;
		logFileManager = new LogFileManager(replicationDirectoryFile);
		Long highestSnapshotId = logFileManager.getHighestId();
		Long sequenceId = Util.getNextServerSequence(highestSnapshotId, serverNumber);
		leases = new Leases(this);
		sequence = new LongSequence(sequenceId, 100);
		threadPool = Executors.newFixedThreadPool(6);
		client = new Client();
	}

	public void resetSequence() {
		Long highestSnapshotId = getHighestId();
		Long sequenceId = Util.getNextServerSequence(highestSnapshotId, serverNumber);
		sequence.set(sequenceId);
	}

	public Long getPreviousId(Long id) {
		return logFileManager.getPreviousId(id);
	}

	public boolean contains(Long id) {
		return logFileManager.contains(id);
	}

	public Long getLowestId() {
		return logFileManager.getLowestId();
	}

	public Long getHighestId() {
		return logFileManager.getHighestId();
	}

	public Record getRecord(EnumSet<Load> load, RecordHeader recordHeader) throws Exception {
		return logFileManager.getRecord(load, recordHeader, documentsBytePool, deletesBytePool);
	}

	public List<RecordHeader> getRecordHeaders(Long snapshotID, int num) throws Exception {
		return logFileManager.getRecordHeaders(snapshotID, num);
	}

	public static HttpTimeout getDefaultHttpTimeout() {
		return new HttpTimeout(1000, 1000);
	}

	public Leases getLeases() {
		return leases;
	}

	public class Client implements TimeoutRunnable {
		private URL url;
		private RestartableTimer timer;

		public void setUrl(URL url) throws InterruptedException {
			this.url = url;
			if (url == null) {
				timer.cancel(false);
			} else {
				timer.runNow();
			}
		}

		public Client() throws Exception {
			timer = new RestartableTimer("replication", this, 5 * 1000, Long.MAX_VALUE);
			setUrl(null);
		}

		public void run(Timeout timeout) {
			if (url == null)
				return;
			try {
				while (true) {
					boolean renewed = renewLease();
					if (renewed) {
						break;
					} else {
						long startid = getHighestId() + 1;
						downloadTransactions(startid, 100);
					}
				}
			} catch (Exception exception) {
				log.info(exception.getMessage());
				runAsOperatingRole();
			}
		}

		private void downloadTransactions(Long startid, int num) throws Exception {
			HttpClient httpClient = new HttpClient();
			HttpMethod httpMethod = new GetMethod(url + "?startid=" + startid + "&num=" + num);
			httpClient.executeMethod(httpMethod);
			int statusCode = httpMethod.getStatusCode();
			if (statusCode == 200) {
				DataInputStream dataInput = new DataInputStream(httpMethod.getResponseBodyAsStream());
				int size = dataInput.readInt();
				for (int x = 0; x < size; x++) {
					ByteBuffers documentsByteBuffers = null;
					ByteBuffers deletesByteBuffers = null;
					try {
						long id = dataInput.readLong();
						String category = dataInput.readUTF();
						long schemaVersion = dataInput.readInt();
						int documentsLength = dataInput.readInt();
						documentsByteBuffers = new ByteBuffers(dataInput, 1024 * 32, documentsLength, documentsBytePool);
						if (documentsByteBuffers.getLength() != documentsLength) {
							throw new Exception("documentsRead not equal to documentsLength");
						}
						int deletesLength = dataInput.readInt();
						deletesByteBuffers = new ByteBuffers(dataInput, 1024 * 4, deletesLength, deletesBytePool);
						if (deletesByteBuffers.getLength() != deletesLength) {
							throw new Exception("deletesRead not equal to deletesLength");
						}
						Data data = new Data(documentsByteBuffers, deletesByteBuffers);
						writeSlaveBatch(id, category, schemaVersion, data);
					} finally {
						if (documentsByteBuffers != null)
							documentsByteBuffers.finished();
						if (deletesByteBuffers != null)
							deletesByteBuffers.finished();
					}
				}
			} else {
				throw new Exception("server error");
			}
		}

		private boolean renewLease() throws Exception {
			HttpParameters parameters = new HttpParameters();
			parameters.add("action", "renewlease");
			Node node = getNode();
			node.transferTo(parameters);
			HttpTimeout httpTimeout = getDefaultHttpTimeout();
			OceanXML response = OceanXML.request(url, parameters, httpTimeout);
			return response.getHeaderBoolean("renewed");
		}
	}

	public boolean renewLease(Node node, long duration) throws Exception {
		if (node.highestLogId != logFileManager.getHighestId()) {
			return false;
		} else {
			leases.renewLease(node, duration);
			return true;
		}
	}

	public static interface NameInterface {
		public void notifyChange();
	}

	// notify name of node status
	public void notifyName() {
		if (nameInterface != null)
			nameInterface.notifyChange();
	}

	public static class Leases {
		private Map<String, Lease> leaseMap = new HashMap<String, Lease>();
		private ReentrantLock lock = new ReentrantLock();
		private Replication replication;

		public Leases(Replication replication) {
			this.replication = replication;
		}

		private void cancelLease(URL url) {
			lock.lock();
			try {
				for (Lease lease : leaseMap.values()) {
					Node node = lease.getNode();
					if (node.getUrl().equals(url)) {
						cancelLease(node.getId());
					}
				}
			} finally {
				lock.unlock();
			}
		}

		private void cancelLease(String id) {
			leaseMap.remove(id);
			replication.notifyName();
		}

		private void renewLease(Node node, long duration) {
			lock.lock();
			try {
				String nodeid = node.getId();
				Lease lease = leaseMap.get(nodeid);
				if (lease == null) {
					lease = new Lease(node);
					leaseMap.put(nodeid, lease);
				}
				lease.renew(duration);
			} finally {
				lock.unlock();
			}
		}

		public List<Node> getNodes() {
			lock.lock();
			try {
				List<Node> nodes = new ArrayList<Node>();
				long currentTime = System.currentTimeMillis();
				for (Map.Entry<String, Lease> entry : leaseMap.entrySet()) {
					Lease lease = entry.getValue();
					Long currentId = replication.getHighestId();
					if (lease.isExpired(currentTime) || !lease.isIdCurrent(currentId)) {
						cancelLease(entry.getKey());
					} else {
						nodes.add(lease.getNode());
					}
				}
				return nodes;
			} finally {
				lock.unlock();
			}
		}

		public static class Lease {
			private long expiration;
			private Node node;

			public Lease(Node node) {
				this.node = node;
			}

			public Node getNode() {
				return node;
			}

			public boolean isIdCurrent(Long id) {
				return ObjectUtils.equals(node.getHighestLogId(), id);
			}

			public boolean isExpired(long currentTime) {
				return currentTime > expiration;
			}

			public void renew(long duration) {
				expiration = duration + System.currentTimeMillis();
			}
		}
	}

	public ByteBufferPool getDeletesByteBuffers() {
		return deletesBytePool;
	}

	public ByteBufferPool getDocumentByteBuffers() {
		return documentsBytePool;
	}

	public void writeSlaveBatch(Long id, String category, long schemaVersion, Data data) throws IOException {
		writeLock.lock();
		try {
			logFileManager.writeRecord(id, category, schemaVersion, data);
			resetSequence();
		} finally {
			writeLock.unlock();
		}
	}

	public void setNodeUrls(Set<URL> newNodeUrls) {
		Set<URL> added = new HashSet<URL>(newNodeUrls);
		added.removeAll(nodeUrls);
		Set<URL> removed = new HashSet<URL>(nodeUrls);
		removed.removeAll(newNodeUrls);
		nodeUrls = newNodeUrls;
		runAsOperatingRole();
	}

	public void runAsOperatingRole() {
		threadPool.submit(new RunAsOperatingRole());
	}

	private class RunAsOperatingRole implements Runnable {
		public void run() {
			try {
				Node master = findVotedMaster(new Timeout(1000 * 60 * 5));
				if (master.getId().equals(id)) {
					operatingRole = "master";
					client.setUrl(null);
					notifyName();
				} else {
					URL url = master.getUrl();
					client.setUrl(url);
					operatingRole = "slave";
				}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public List<Node> getNodes() {
		try {
			MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.NONE);
			HttpTimeout defaultHttpTimeout = getDefaultHttpTimeout();
			HttpParameters parameters = new HttpParameters("action", "getnode");
			List<OceanXML> responses = multicastHttp.execute(nodeUrls, parameters, defaultHttpTimeout);
			List<Node> nodes = new ArrayList<Node>();
			for (OceanXML response : responses) {
				Node node = null;
				nodes.add(node);
			}
			return nodes;
		} catch (Throwable throwable) {
			throw new RuntimeException(throwable);
		}
	}

	private Node findVotedMaster(Timeout timeout) throws Throwable, InterruptedException, TimeoutException {
		while (timeout.isOK()) {
			Node node = findVotedMaster();
			if (node != null)
				return node;
		}
		throw new TimeoutException(timeout);
	}

	public String getOperatingRole() {
		return operatingRole;
	}

	private Node findVotedMaster() throws InterruptedException, Throwable {
		MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.NONE);
		HttpTimeout defaultHttpTimeout = getDefaultHttpTimeout();
		HttpParameters parameters = new HttpParameters("action", "getchosenmaster");
		List<OceanXML> responses = multicastHttp.execute(nodeUrls, parameters, defaultHttpTimeout);
		if (isMajority(responses.size())) {
			Map<String, Node> chosenMasterMap = new HashMap<String, Node>();
			for (OceanXML response : responses) {
				Node node = new Node(response.getFirstInBody());
				chosenMasterMap.put(node.getChosenMasterId(), node);
			}
			if (chosenMasterMap.size() == 1) {
				return chosenMasterMap.values().iterator().next();
			}
		}
		return null;
	}

	private boolean isMajority(int size) {
		if ((size / 2) > nodeUrls.size()) {
			return true;
		} else
			return false;
	}

	public Node getNode() {
		String role = getOperatingRole();
		Long highestLogID = sequence.get();
		String chosenMasterID = null;
		if (chosenMaster != null) {
			chosenMasterID = chosenMaster.getId();
		}
		Node node = new Node(id, url, chosenMasterID, role, highestLogID);
		return node;
	}

	public boolean isChosenMaster() {
		Node node = getChosenMaster();
		if (node == null)
			return false;
		return StringUtils.equals(node.getId(), id);
	}

	public Node getChosenMaster() {
		if (chosenMaster == null) {
			chosenMaster = chooseMaster(getNodes());
		}
		return chosenMaster;
	}

	private static Node chooseMaster(List<Node> nodes) {
		int masterCount = getMasterCount(nodes);
		if (masterCount > 1)
			return null;
		else if (masterCount == 1) {
			return getMaster(nodes);
		} else if (masterCount == 0) {
			try {
				ReflectColumns reflectColumns = new ReflectColumns();
				reflectColumns.add("highestCellTransactionIDDatabase", SortReflect.method("getHighestLogID"), SorterConstants.DESCENDING_ORDER);
				reflectColumns.add("id", SortReflect.method("getId"), SorterConstants.ASCENDING_ORDER);
				Sorter sorter = Sorter.createReflect(reflectColumns, nodes, SorterConstants.NULLS_ARE_LEAST);
				List<Node> sortedNodes = sorter.sortReflect();
				return sortedNodes.get(0);
			} catch (MethodException methodException) {
				throw new RuntimeException(methodException.getCause());
			}
		}
		throw new RuntimeException("should not have reached");
	}

	private static Node getMaster(Collection<Node> nodes) {
		for (Node node : nodes) {
			String role = node.getOperatingRole();
			if (StringUtils.equals(role, "master"))
				return node;
		}
		return null;
	}

	private static int getMasterCount(Collection<Node> nodes) {
		int count = 0;
		for (Node node : nodes) {
			String role = node.getOperatingRole();
			if (StringUtils.equals(role, "master"))
				count++;
		}
		return count;
	}

	public static class Node implements CElement {
		private String id;
		private URL url;
		private String chosenMasterId;
		private String operatingRole;
		private Long highestLogId;

		public Node(Element element) throws Exception {
			XMLUtil.reflectToObject(this, element);
		}

		public void transferTo(HttpParameters parameters) {
			parameters.add("id", id);
			parameters.add("url", url);
			parameters.add("chosenMasterID", chosenMasterId);
			parameters.add("operatingRole", operatingRole);
			parameters.add("highestLogID", highestLogId);
		}

		public Node(HttpParameters parameters) throws MalformedURLException {
			id = parameters.get("id");
			url = parameters.getUrl("url");
			chosenMasterId = parameters.get("chosenMasterID");
			operatingRole = parameters.get("operatingRole");
			highestLogId = parameters.getLong("highestLogID");
		}

		public Node(String id, URL url, String chosenMasterId, String operatingRole, Long highestLogID) {
			this.id = id;
			this.url = url;
			this.chosenMasterId = chosenMasterId;
			this.operatingRole = operatingRole;
			this.highestLogId = highestLogID;
		}

		public Long getHighestLogId() {
			return highestLogId;
		}

		public Element toElement() throws Exception {
			return XMLUtil.reflectToElement("node", this);
		}

		public String getId() {
			return id;
		}

		public URL getUrl() {
			return url;
		}

		public String getChosenMasterId() {
			return chosenMasterId;
		}

		public String getOperatingRole() {
			return operatingRole;
		}

		public boolean equals(Object object) {
			if (!(object instanceof Node))
				return false;
			Node other = (Node) object;
			EqualsBuilder equalsBuilder = new EqualsBuilder();
			equalsBuilder.append(id, other.id);
			equalsBuilder.append(url, other.url);
			equalsBuilder.append(chosenMasterId, other.chosenMasterId);
			// equalsBuilder.append(operatingRole, other.operatingRole);
			return equalsBuilder.isEquals();
		}

		public int hashCode() {
			HashCodeBuilder hashCodeBuilder = new HashCodeBuilder(555, 121);
			hashCodeBuilder.append(id);
			hashCodeBuilder.append(url);
			hashCodeBuilder.append(chosenMasterId);
			// hashCodeBuilder.append(operatingRole);
			return hashCodeBuilder.toHashCode();
		}
	}

	public static class ReplicationException extends Exception {
		public ReplicationException(String message, Throwable throwable) {
			super(message, throwable);
		}
	}

	public Long getNextId() {
		return sequence.getAndIncrement();
	}

	public SlaveBatch getBatchGreaterOrEqualTo(String category, Long id) throws Exception {
		EnumSet<Load> load = EnumSet.of(Load.DOCUMENTS, Load.DELETES);
		return getBatchGreaterOrEqualTo(load, category, id);
	}

	public List<SlaveBatch> getBatchesGreaterOrEqualTo(EnumSet<Load> load, String category, Long id) throws Exception {
		List<SlaveBatch> slaveBatches = new ArrayList<SlaveBatch>();
		Long currentID = id;
		for (;;) {
			SlaveBatch slaveBatch = getBatchGreaterOrEqualTo(load, category, id);
			if (slaveBatch != null) {
				slaveBatches.add(slaveBatch);
				currentID = slaveBatch.getId() + 1;
			} else {
				break;
			}
		}
		return slaveBatches;
	}

	public SlaveBatch getBatchGreaterOrEqualTo(EnumSet<Load> load, String category, Long id) throws Exception {
		Record record = null;
		try {
			record = getRecordGreaterOrEqualTo(load, category, id);
			Data data = record.getData();
			IndexDocuments indexDocuments = null;
			RAMDirectory ramDirectory = null;
			Deletes deletes = null;
			if (load.contains(Load.DOCUMENTS)) {
				InputStream inputStream = data.getDocumentsByteBuffers().getInputStream();
				ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
				Object unserialized = objectInputStream.readObject();
				if (unserialized instanceof SerializableRamDirectory) {
					SerializableRamDirectory serializableRamDirectory = (SerializableRamDirectory) unserialized;
					ramDirectory = serializableRamDirectory.getRamDirectory();
				} else if (unserialized instanceof SerializableBatchDocuments) {
					SerializableBatchDocuments serializableBatchDocuments = (SerializableBatchDocuments) unserialized;
					indexDocuments = serializableBatchDocuments.getDocuments();
				}
			}
			if (load.contains(Load.DELETES)) {
				InputStream inputStream = data.getDeletesByteBuffers().getInputStream();
				ObjectInputStream deletesObjectInputStream = new ObjectInputStream(inputStream);
				SerializableBatchDeletes serializableBatchDeletes = (SerializableBatchDeletes) deletesObjectInputStream.readObject();
				deletes = serializableBatchDeletes.getDeletes();
			}
			long schemaVersion = record.getSchemaVersion();
			IndexSchema indexSchema = schemaLoader.getIndexSchema(schemaVersion, new Timeout(1000));
			IndexSchema.Category schemaCategory = indexSchema.getCategory(record.getCategory());
			if (ramDirectory != null) {
				return new SlaveBatch(category, id, schemaCategory, ramDirectory, deletes);
			} else {
				return new SlaveBatch(category, id, schemaCategory, indexDocuments, deletes);
			}
		} catch (ClassNotFoundException classNotFoundException) {
			IOException ioException = new IOException(classNotFoundException.getMessage());
			ioException.initCause(classNotFoundException);
			throw ioException;
		} finally {
			if (record != null) {
				Data data = record.getData();
				if (data != null)
					data.finished();
			}
		}
	}

	public Record getRecordGreaterOrEqualTo(Long id) throws IOException {
		EnumSet<Load> load = EnumSet.of(Load.DOCUMENTS, Load.DELETES);
		return logFileManager.getRecordGreaterOrEqualTo(load, id, documentsBytePool, deletesBytePool);
	}

	public Record getRecordGreaterOrEqualTo(EnumSet<Load> load, String category, Long id) throws IOException {
		return logFileManager.getRecordGreaterOrEqualTo(load, category, id, documentsBytePool, deletesBytePool);
	}

	public void writeMasterBatch(final Long id, final Long previousId, final MasterBatch masterBatch) throws Exception {
		final String category = masterBatch.getCategoryName();
		Serializable serializableDocuments = masterBatch.getSerializableBatchDocuments(id);
		Serializable serializableDeletes = masterBatch.getSerializableBatchDeletes(id);
		final ByteArrayOutputStream documentsByteOutput = new ByteArrayOutputStream(getDocumentByteBuffers());
		final ByteArrayOutputStream deletesByteOutput = new ByteArrayOutputStream(getDeletesByteBuffers());
		try {
			ObjectOutputStream documentsObjectOutputStream = new ObjectOutputStream(documentsByteOutput);
			documentsObjectOutputStream.writeObject(serializableDocuments);
			documentsObjectOutputStream.close();
			ObjectOutputStream deletesObjectOutputStream = new ObjectOutputStream(deletesByteOutput);
			deletesObjectOutputStream.writeObject(serializableDeletes);
			deletesObjectOutputStream.close();
			final int documentsBytesLength = documentsByteOutput.size();
			final int deletesBytesLength = deletesByteOutput.size();
			ByteBuffers documentsByteBuffers = documentsByteOutput.getByteBuffers();
			ByteBuffers deletesByteBuffers = deletesByteOutput.getByteBuffers();
			final Data data = new Data(documentsByteBuffers, deletesByteBuffers);
			writeLock.lock();
			try {
				final long schemaVersion = masterBatch.getSchemaVersion();
				// TODO: add commit to logFileManager which adds record to the map if successful transaction
				// TODO: multithread writerecord with http multicast
				//RecordHeader recordHeader = logFileManager.writeRecord(id, category, schemaVersion, data);
				Future<RecordHeader> writeRecordFuture = threadPool.submit(new Callable<RecordHeader>() {
					public RecordHeader call() throws Exception {
						return logFileManager.writeRecord(id, category, schemaVersion, data);
					}
				});
				List<URL> urls = new ArrayList<URL>();
				HttpParameters httpParameters = new HttpParameters();
				httpParameters.add("action", "writetransaction");
				httpParameters.add("id", id);
				httpParameters.add("category", category);
				httpParameters.add("schemaversion", schemaVersion);
				httpParameters.add("previousid", previousId);
				List<Node> nodes = leases.getNodes();
				for (Node node : nodes) {
					urls.add(new URL(node.getUrl().toString() + "?" + httpParameters.toEncodedQueryString()));
				}
				MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.ALL);
				HttpTimeout httpTimeout = new HttpTimeout(1000, 3 * 1000);
				try {
					List<OceanXML> responses = multicastHttp.executePost(urls, httpTimeout, new RequestEntity() {
						public void writeRequest(OutputStream output) throws IOException {
							DataOutputStream dataOutput = new DataOutputStream(output);
							dataOutput.writeInt(documentsBytesLength);
							documentsByteOutput.writeTo(dataOutput);
							dataOutput.writeInt(deletesBytesLength);
							deletesByteOutput.writeTo(dataOutput);
						}

						public long getContentLength() {
							return documentsBytesLength + deletesBytesLength;
						}

						public String getContentType() {
							return "binary/octet";
						}

						public boolean isRepeatable() {
							return true;
						}
					});
					for (OceanXML response : responses) {
						if (response.getHeaderBoolean("cancellease")) {
							String serverid = response.getHeaderValue("serverid");
							leases.cancelLease(serverid);
						} else if (response.getHeaderBoolean("renewlease")) {
							Node node = new Node(response.getFirstInBody());
							long duration = response.getHeaderLong("leaseduration");
							leases.renewLease(node, duration);
						}
					}
				} catch (AllExceptions allExceptions) {
					List<HttpRequestException> exceptions = allExceptions.getExceptions();
					if (exceptions.size() == nodes.size()) {
						// nothing worked
					}
					for (HttpRequestException requestException : exceptions) {
						URL url = requestException.getUrl();
						leases.cancelLease(url);
					}
					// TODO: throw exception
				} catch (Exception exception) {
					log.log(Level.SEVERE, "", exception);
					throw exception;
				}
				RecordHeader recordHeader = writeRecordFuture.get();
				logFileManager.commit(recordHeader);
				// TODO: commit recordheader
				// TODO: what if local write failed?
			} finally {
				writeLock.unlock();
			}
		} finally {
			documentsByteOutput.close();
			deletesByteOutput.close();
		}
	}
}
